package com.lagou.boot;

import com.lagou.client.RPCConsumer;
import com.lagou.entity.RpcRequest;
import com.lagou.handler.CommonHandler;
import com.lagou.service.IUserService;
import com.lagou.zookeeper.DataChange;
import com.lagou.zookeeper.ZookeeperUtil;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ConsumerBoot {

    //参数定义
    private static final String PROVIDER_NAME = "UserService#sayHello#";


    private static Map<String,RPCConsumer> rpcConsumerMap = new HashMap<>();

    private static ZookeeperUtil zookeeperUtil =  new ZookeeperUtil("192.168.147.180",2181);


    private static Map<String,Long> responseTimeMap = new HashMap<>();

    public static void main(String[] args) throws InterruptedException {


        //初始化服务器列表
        initServerList();

        //监听服务器列表变化
        watchServerList();

        //定时请求netty服务端
        requestToNettyServer();

        //定时上报netty服务端响应时间
        reportResponseTime();


    }

    private static void reportResponseTime(){
        //每5s上报一次
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    zookeeperUtil.writeResponseTimeMap(responseTimeMap);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }, 0, 5, TimeUnit.SECONDS);
    }

    private static RPCConsumer selectNettyServer(){
        Long min = 100000000L;
        String server = "";
        Map<String,Long> serverResponseTime = zookeeperUtil.readResponseTimeMap();
        //如果zookeep还没有服务器的响应时间信息，那就先在当前服务器中选一个
        if (serverResponseTime == null || serverResponseTime.size() == 0){
            for (String serverIp : rpcConsumerMap.keySet()){
                return rpcConsumerMap.get(serverIp);
            }
        }
        for (String key : serverResponseTime.keySet()){
            //如果服务器已经下线，剔除掉
            if (!rpcConsumerMap.containsKey(key)){
                continue;
            }
            Long responseTime = serverResponseTime.get(key);
            if (responseTime < min){
                server = key;
                min = responseTime;
            }
        }
        return rpcConsumerMap.get(server);
    }

    private static void requestToNettyServer(){
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    RPCConsumer rpcConsumer = selectNettyServer();
                    if (rpcConsumer == null){
                        System.out.println("还没有netty服务器");
                        return;
                    }
                    System.out.println("选取服务器===>" + rpcConsumer.getHost() + ":" + rpcConsumer.getPort());
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setClassName("com.coding365.rpcprovider.service.UserServiceImpl");
                    rpcRequest.setMethodName("sayHello");
                    String msg = "are you ok?";
                    rpcRequest.setParameters(new Object[]{msg});
                    rpcRequest.setParameterTypes(new Class<?>[]{String.class});
                    rpcConsumer.call(rpcRequest);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }, 0, 4, TimeUnit.SECONDS);
    }



    private static RPCConsumer generateRPCConsumer(String server){
        CommonHandler commonHandler = new CommonHandler() {
            @Override
            protected void readData(RpcRequest result) {
                try {
                    Long requestTimeStamp = result.getRequestTimeStamp();
                    Long nowTimeStamp = System.currentTimeMillis();
                    //请求响应时间
                    Long responseTime = nowTimeStamp - requestTimeStamp;
                    System.out.println("responseTime:" + responseTime);
                    responseTimeMap.put(server,responseTime);
                }catch (Exception e){
                    e.printStackTrace();
                }

            }
        };

        String[] parts = server.split(":");
        String host = parts[0];
        String portStr = parts[1];
        Integer port = Integer.valueOf(portStr);

        RPCConsumer rpcConsumer = new RPCConsumer(host,port,commonHandler);

        return rpcConsumer;
    }

    private static void initServerList(){
        List<String> serverList = zookeeperUtil.readServerList();
        if (serverList != null && serverList.size() > 0){
            for (String server : serverList){
                RPCConsumer rpcConsumer = generateRPCConsumer(server);
                rpcConsumerMap.put(server,rpcConsumer);
                responseTimeMap.put(server,0L);
            }
        }
    }

    private static void watchServerList(){
        zookeeperUtil.watchServerList( new DataChange() {
            @Override
            public void dataChange(String s, List<String> serverList) {
                //变化的服务器列表
                Map<String,String> serverMap = new HashMap();
                for (String server : serverList){
                    serverMap.put(server,server);
                }

                //处理删除的服务器
                List<String> serverToDelete = new ArrayList<>();
                for (String key : rpcConsumerMap.keySet()){
                    String server = serverMap.get(key);
                    if (server == null || "".equals(server)){
                        RPCConsumer rpcConsumer = rpcConsumerMap.get(key);
                        rpcConsumer.close();
                        serverToDelete.add(key);
                    }
                }

                for (String delServer : serverToDelete){
                    rpcConsumerMap.remove(delServer);
                }

                //处理增加服务器
                for (String server : serverMap.keySet()){
                    if (null == rpcConsumerMap.get(server)){
                        RPCConsumer rpcConsumer = generateRPCConsumer(server);
                        rpcConsumerMap.put(server,rpcConsumer);
                    }
                }

            }
        });
    }
}
